package cn.edu.xmu.datamining.tangzk.mralgos.kmeans;

import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class KMeansReducer extends
		Reducer<ClusterCentroidWritable, SampleWritable, IntWritable, Text> {

	private float threshhold = 1e-2f;

	protected void setup(
			org.apache.hadoop.mapreduce.Reducer<ClusterCentroidWritable, SampleWritable, IntWritable, Text>.Context context)
			throws java.io.IOException, InterruptedException {
		Configuration conf = context.getConfiguration();
		threshhold = conf.getFloat(KMeansDriver.CENTROID_DIFF_THRESHHOLD,
				threshhold);
	};

	protected void reduce(
			ClusterCentroidWritable key,
			java.lang.Iterable<SampleWritable> values,
			org.apache.hadoop.mapreduce.Reducer<ClusterCentroidWritable, SampleWritable, IntWritable, Text>.Context context)
			throws java.io.IOException, InterruptedException {
		Iterator<SampleWritable> valIter = values.iterator();
		SampleWritable mean = null;
		boolean flag = false; // sample exists or not
		SampleWritable sample = null;
		while (valIter.hasNext()) {
			sample = valIter.next();
			if (key.isCenter()) {
				mean = key.getSample();
			} else {
				flag = true;
				break;
			}
		}
		if (flag) {
			// note: sample will be changed in SampleWritable.getMean(...) when
			// iterating the <key,value> pair
			mean = SampleWritable.getMean(sample.clone(), values);
			// determine whether the centroid is changed.
			double diff = key.getSample().getDistance(mean);
			if (diff > threshhold) {
				context.getCounter(KMeansDriver.USER_DEFINED_GROUP,
						KMeansDriver.CNT_CENTROID_DIFF).increment(1);
			}
		}
		context.write(new IntWritable(key.getC()), new Text(mean.toString()));
	};

}
